package usbipclientsrv

import (
	log "github.com/sirupsen/logrus"
	"go.uber.org/atomic"
	"net"
	"sync"
	"time"
	"usbipproxy/common"
	"usbipproxy/domain/service"
	"usbipproxy/pb"
)

type UsbipCliSession struct {
	conn          net.Conn
	outAttachData chan []byte
	closed        atomic.Bool
	buf           []byte
	clientID      string
	msgNotifySvc  service.MsgNotifyService
	sessId        string
	busId         string
	vendorId      string
	logger        *log.Entry
	notifyConnId  int64
	wg            sync.WaitGroup
}

func NewUsbipCliSession(clientID, busId, vendorId, sessId string, notifyConnId int64, conn net.Conn, msgNotifySvc service.MsgNotifyService) *UsbipCliSession {
	return &UsbipCliSession{
		clientID:      clientID,
		conn:          conn,
		outAttachData: make(chan []byte, 60000),
		buf:           make([]byte, 64*1024),
		msgNotifySvc:  msgNotifySvc,
		sessId:        sessId,
		busId:         busId,
		vendorId:      vendorId,
		notifyConnId:  notifyConnId,
		logger:        log.WithField("bus_id", busId).WithField("sess_id", sessId).WithField("client_id", clientID),
	}
}

func (t *UsbipCliSession) SendHeartbeat() {
	if t.closed.Load() {
		return
	}
	t.msgNotifySvc.SendMsg(&pb.ProxyMsg{
		MsgType: pb.MsgType_HEARTBEAT,
		Base: &pb.BaseParam{
			//Timestamp: time.Now().UnixNano(),
			ToId: t.notifyConnId,
		},
		Param: &pb.ProxyMsg_HeartbeatParam{HeartbeatParam: &pb.HeartbeatParam{
			BusId:  t.busId,
			SessId: t.sessId,
		}},
	})
}

func (t *UsbipCliSession) FlushAllOutMsg() {
	for {
		select {
		case <-t.outAttachData:
		default:
			return
		}
	}
}

func (t *UsbipCliSession) Stop() {
	if !t.closed.CompareAndSwap(false, true) {
		return
	}
	t.conn.Close()
	t.logger.Infof("UsbipCliSession sessId:%v wait", t.sessId)
	t.wg.Wait()
	t.logger.Infof("UsbipCliSession sessId:%v over and stop", t.sessId)
}

func (t *UsbipCliSession) SendMsg(msg []byte) error {
	/*
		if t.closed.Load() {
			log.Infof("UsbipCliSession have been closed")
			return nil
		}
	*/
	select {
	case t.outAttachData <- msg:
	default:
		t.logger.Infof("UsbipCliSession send msg fail, channel is full")
	}

	return nil
}

func (t *UsbipCliSession) writeBinaryData(data []byte) error {
	// 写入实际数据
	writeTimeout := 30 * time.Second
	deadline := time.Now().Add(writeTimeout)
	if err := t.conn.SetWriteDeadline(deadline); err != nil {
		t.logger.Warnf("UsbipCliSession SetWriteDeadline fail: %v")
		return err
	}
	n, err := common.BlockingWrite(t.conn, data)
	if err != nil {
		t.logger.Warnf("writeBinaryData fail: %v", err)
		return err
	}

	if n != len(data) {
		t.logger.Warnf("UsbipCliSession writeBinaryData failed:%v:%v", len(data), n)
	}

	t.logger.Debugf("UsbipCliSession writeBinaryData n:%d len(data):%v", n, len(data))
	return nil
}

func (t *UsbipCliSession) handleConnection() {

	termMsg := &pb.ProxyMsg{
		MsgType: pb.MsgType_TO_SERVER_ATTACH,
		/*Param: &pb.ProxyMsg_ServerBeAttachedParam{ServerBeAttachedParam: &pb.ServerBeAttachedParam{
			SessId: t.sessId,
			BusId:  t.busId,
		}},
		*/
		Base: &pb.BaseParam{
			ToId: t.notifyConnId,
		},
		Param: &pb.ProxyMsg_ToServerAttachParam{ToServerAttachParam: &pb.ToServerAttachParam{
			SessId:   t.sessId,
			BusId:    t.busId,
			VendorId: t.vendorId,
		}},
	}
	t.msgNotifySvc.SendMsg(termMsg)

	// 为数据分发启动单独的goroutine
	t.wg.Add(1)
	go func() {
		defer t.wg.Done()
		defer func() {
			termMsg1 := &pb.ProxyMsg{
				MsgType: pb.MsgType_TO_SERVER_ATTACH_OVER,
				Base: &pb.BaseParam{
					ToId: t.notifyConnId,
				},
				Param: &pb.ProxyMsg_ToServerAttachOverParam{ToServerAttachOverParam: &pb.ToServerAttachOverParam{
					SessId:   t.sessId,
					BusId:    t.busId,
					VendorId: t.vendorId,
				}},
			}
			t.msgNotifySvc.SendMsg(termMsg1)
			t.Stop()
			t.FlushAllOutMsg()
			t.logger.Infof("UsbipCliSession handleConnection end")
			//t.removeClient(clientID)
		}()
		for msg := range t.outAttachData {
			if len(msg) == 0 {
				break
			}
			if err := t.writeBinaryData(msg); err != nil {
				t.logger.Infof("UsbipCliSession handleConnection Write error %v\n", err)
				return
			}

			t.logger.Debugf("UsbipCliSession handleConnection Write len(data):%v", len(msg))
		}
	}()

	// 主循环：读取客户端数据
	for {
		// 从对象池获取缓冲区
		n, err := t.conn.Read(t.buf)
		if err != nil {
			t.logger.Infof("UsbipCliSession handleConnection Read error : %v", err)
			//链接断了
			break
		}

		if n > 0 {
			// 复制实际读取的数据
			t.logger.Debugf("UsbipCliSession handleConnection read %d", n)
			data := make([]byte, n)
			copy(data, t.buf[:n])
			//log.Debugf("UsbipCliSession handleConnection data %v st", data, string(data))
			t.msgNotifySvc.SendMsg(&pb.ProxyMsg{
				MsgType: pb.MsgType_TO_SERVER_ATTACH_DATA,
				Base: &pb.BaseParam{
					ToId: t.notifyConnId,
				},
				Param: &pb.ProxyMsg_ToServerAttachDataParam{ToServerAttachDataParam: &pb.ToServerAttachDataParam{
					BusId:  t.busId,
					SessId: t.sessId,
					Data:   data,
				}},
			})
		}
	}

	t.SendMsg(nil)
}
